package com.movitech.mobile.mq;

import com.alibaba.fastjson.JSONObject;
import com.movitech.utils.LoggerManager;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * Copyright: Copyright (c) 2021 Asiainfo
 *
 * @ClassName: com.actionsoft.apps.rabbitmq.util.RabbitmqConllectionUtil
 * @Description: 该类的功能描述：RabbitMQ数据连接
 * @version: v1.0.0
 * @author: xdq
 * @date: 2021/6/2 9:51 上午
 */
public class RabbitmqConllection {
    private static LoggerManager log = LoggerManager.get(RabbitmqConllection.class);
    private String mqUrl;
    private String mqUser;
    private String mqPwd;
    private String mqNPort;
    static Map<String, RabbitmqConllection> connectionMap = new HashMap<>();
    Connection connection = null;

    public static synchronized RabbitmqConllection instance(String url, String port, String user, String pwd) {
        String key = url + port + user + pwd;
        if (connectionMap.get(url + port + user + pwd) == null) {
            RabbitmqConllection conllection = new RabbitmqConllection();
            conllection.mqUrl = url;
            conllection.mqUser = user;
            conllection.mqPwd = pwd;
            conllection.mqNPort = port;
            connectionMap.put(key, conllection);
        }
        return connectionMap.get(url + port + user + pwd);
    }

    /**
     * 根据rabbitMqName获取一个连接，使用完记得要自己关闭连接 conn.close()
     */
    public synchronized Connection getConnection() {
        if (connection != null && connection.isOpen()) {
            return connection;
        }
        try {
            ConnectionFactory connectionFactory = new ConnectionFactory();
            //virtual_host，类似Mysql中数据库
            connectionFactory.setVirtualHost("/");
            //用户名
            connectionFactory.setUsername(mqUser);
            //密码
            connectionFactory.setPassword(mqPwd);
            //设置自动重新获取连接为true
            connectionFactory.setAutomaticRecoveryEnabled(true);
            //设置自动重新获取网络时间为10000ms=10s
            connectionFactory.setNetworkRecoveryInterval(10000);

            String[] urlList = mqUrl.split(";");
            String[] portList = mqNPort.split(";");

            if (urlList.length != portList.length) {
                throw new RuntimeException("地址数量与端口数量不匹配");
            }


            Address[] addresses = new Address[urlList.length];

            for (int i = 0; i < urlList.length; i++) {
                addresses[i] = new Address(urlList[i], Integer.parseInt(portList[i]));
            }
            connection = connectionFactory.newConnection(addresses);
        } catch (Exception e) {
            log.debug("创建MQ连接失败", e);
        } finally {
            return connection;
        }
    }

    /**
     * 消费者
     *
     * @param queueName
     * @param consumer
     * @return
     * @author luodengzhong
     * @version 1.0.0
     * @createTime 2021/8/16
     * @copyright Copyright ©️ 2021 北京魔马科技
     */
    public void consumer(String queueName, RabbitmqConsumer consumer) {
        try {
            // 获取到连接
            Connection connection = this.getConnection();
            // 获取通道
            Channel channel = connection.createChannel();
            // 声明队列
            channel.queueDeclare(queueName, true, false, false, null);

            channel.basicConsume(queueName, false, new DefaultConsumer(channel) {
                @Override
                public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                    // body 即消息体
                    String data = new String(body, "utf-8");
                    boolean r = false;
                    try {
                        r = consumer.consume(queueName, data);
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    if (r) {
                       //这里的false表示不自动ack，为true表示自动ack，但是自动ack会有消息丢失的潜在问题
                        channel.basicAck(envelope.getDeliveryTag(), false);
                    }
                }
            });
        } catch (Exception e) {
            consumer.error(queueName, e.getMessage(), e);
        }
    }

    /**
     * 生产者
     *
     * @param
     * @return
     * @author luodengzhong
     * @version 1.0.0
     * @createTime 2021/8/16
     * @copyright Copyright ©️ 2021 北京魔马科技
     */
    public void producer(String queueName, RabbitmqProducer errorBack, String jsonStr) {
        Connection connection = null;
        Channel channel = null;
        try {
            //获取连接
            connection = getConnection();
            //从连接中获取一个通道
            channel = connection.createChannel();

            //声明队列
            channel.queueDeclare(queueName, true, false, false, null);
            String message = jsonStr;
            //发送消息
            channel.basicPublish("", queueName, null, message.getBytes("utf-8"));
        } catch (Exception e) {
            //消息发送失败回调
            if (errorBack != null) {
                errorBack.error(jsonStr, e);
            }
            throw new RuntimeException("消息推送失败：" + jsonStr, e);
        } finally {
            if (channel != null) {
                try {
                    channel.close();
                } catch (Exception e) {
                    log.error(e.getMessage(), e);
                }
                channel = null;
                connection = null;
            }

        }
    }

    public void closeConnection() {
        try {
            this.connection.close();
        } catch (IOException e) {
            log.error("关闭连接失败" + e.getMessage(), e);
        }
        this.connection = null;
    }
}
